package com.seari.client.kafka;

import java.io.IOException;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.seari.bean.IorBean;
import com.seari.bean.ResultTrain;
import com.seari.bean.TrainInfo;
import com.seari.bean.WindowInfo;
import com.seari.controller.ATSWebsocketController;
import com.seari.main.StartupClass;
import com.seari.pojo.ATSMessageBean;
import com.seari.service.AppCacheService;
import com.seari.service.TrainService;
import com.seari.service.TrainServiceImpl;
import com.seari.utils.SpringContainer;

public class KafkaWorker implements Runnable
{
	protected static Logger logger = LoggerFactory.getLogger(KafkaWorker.class);
	public KafkaConsumer<?, ?> consumer;
	public boolean terminateFlag = false;
	private static final long MAX_TIME_GAP = 2 * 60 * 1000; //最大信息有效间隔：2分钟
	private static DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH-mm-ss");
	public final static ObjectMapper MAPPER = new ObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
	private String workerName ="";
	
	public KafkaWorker(List<String> topics,String threadName)
	{
		Properties props = new Properties();
        props.put("bootstrap.servers", StartupClass.config.getProperty("kafka_bootstrap_servers"));
        props.put("group.id", StartupClass.config.getProperty("kafka_consumer_groupid"));
        props.put("enable.auto.commit", Boolean.parseBoolean(StartupClass.config.getProperty("kafka_consumer_enable_auto_commit","true")));
        props.put("auto.commit.interval.ms", Integer.parseInt(StartupClass.config.getProperty("kafka_consumer_auto_commit_interval","1000")));
        props.put("auto.offset.reset", StartupClass.config.getProperty("kafka_consumer_auto_offset_reset","latest"));
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<String, String>(props);
        this.consumer.subscribe(topics);
        this.setWorkerName(threadName);
	}
	@Override
	public void run() 
	{
		Thread.currentThread().setName(workerName);
		while(!terminateFlag)
		{
			try
			{
				ConsumerRecords<?, ?> records = consumer.poll(Duration.ofMillis(300));
				for (ConsumerRecord<?, ?> record : records) 
				{
					Optional<?> kafkaMessage = Optional.ofNullable(record.value());
					if (kafkaMessage.isPresent())
					{
						Object message = kafkaMessage.get();
						//logger.info("----------------- record =" + record);
						//logger.info("------------------ message =" + message);
						try
						{
							if(StringUtils.isNotBlank(message.toString()))
							{
								long currentTime = System.currentTimeMillis();
								String messageStr = message.toString();
								JsonNode jsonNode = MAPPER.readTree(messageStr);
								jsonNode.get("msgHead").get("sendTime").asText();
								long sendTime = LocalDateTime.parse(jsonNode.get("msgHead").get("sendTime").asText(),dtf).toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
								if(currentTime - sendTime > MAX_TIME_GAP) 
								{
									//丢弃2分钟前滞后数据
									logger.warn("message lag too long:{}",messageStr);
									return;
								}
								String lineId = jsonNode.get("msgHead").get("lineId").asText();
								if(lineId.length() == 1) //线路号只有1位时，在前方补0
								{
									lineId = "0" + lineId;
								}
								int msgId = jsonNode.get("msgHead").get("msgId").asInt(0);
								//根据msgId决定处理方式
								if(msgId == KafkaMessageHelper.MESSAGE_ALL)
								{
									//do nothing
									logger.info("receive kafka all message:{}",messageStr);
								}else if(msgId == KafkaMessageHelper.MESSAGE_UPDATE) {
									TrainInfo trainInfo = MAPPER.treeToValue(jsonNode.get("msgBody"),TrainInfo.class);
									trainInfo.setUpdateTime(sendTime);
									trainInfo.setLineId(lineId);
									//System.out.println(trainInfo);
									convertTrainInfo(trainInfo);
								}else if(msgId == KafkaMessageHelper.MESSAGE_REMOVE) {
									//do nothing
									logger.info("receive kafka remove message:{}",messageStr);
									TrainInfo trainInfo = MAPPER.treeToValue(jsonNode.get("msgBody"),TrainInfo.class);
									trainInfo.setUpdateTime(sendTime);
									trainInfo.setLineId(lineId);
									removeTrain(trainInfo);
								}else if(msgId == KafkaMessageHelper.MESSAGE_ALIVE_STATUS) {
									//do nothing
									logger.info("receive kafka alive message:{}",messageStr);
								}else {
									//do nothing
								}
							}
						} catch (Exception e)
						{
							// TODO: handle exception
							e.printStackTrace();
						}
					}
				}
			} catch (Exception e)
			{
				logger.error("kafka worker error:{}, consumer status:{}",e.getMessage(),consumer.toString());
				e.printStackTrace();
				try
				{
					//consumer产生错误时休眠10秒
					TimeUnit.SECONDS.sleep(10);
				} catch (InterruptedException e1)
				{
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
			}
		}
	}
	
	public ResultTrain convertTrainInfo(TrainInfo originInfo)
	{
		ResultTrain resultTrain = null;
		String lineId = originInfo.getLineId();
		if(lineId.equals(KafkaMessageHelper.LINE_03))
		{
			//3、4号线共线段特殊处理，列车组编号以4起始时认定为4号线
			if (originInfo.getTrainGroupId().startsWith("4"))
			{
				lineId = KafkaMessageHelper.LINE_04;
				originInfo.setLineId(lineId);
			}
		}
		if(TrainServiceImpl.trainMap.get(lineId) == null)//线路不存在时进行创建线路及对应的车辆信息
		{
			if(TrainServiceImpl.trainMap.get(lineId) == null) 
			{
				TrainServiceImpl.trainMap.put(lineId, new ConcurrentHashMap<>());
				TrainServiceImpl.trainMap.get(lineId).put(originInfo.getTrainGroupId(), process(originInfo, resultTrain));
			}
		}else {
			//线路存在时进行车辆信息是否存在判断
			ConcurrentHashMap<String, ResultTrain> trainRefMap = TrainServiceImpl.trainMap.get(lineId);
			//不存在时转换至ResultTrain并添加到MAP中
			if(trainRefMap.get(originInfo.getTrainGroupId()) == null)
			{
				trainRefMap.put(originInfo.getTrainGroupId(), process(originInfo, resultTrain));
			}
			else {
				//存在时先获取既有车辆信息进行更新
				process(originInfo, trainRefMap.get(originInfo.getTrainGroupId()));
			}
		}
		return resultTrain;
	}
	
	private ResultTrain process(TrainInfo originInfo,ResultTrain originTrain)
	{
		if(originTrain == null) 
		{
			originTrain = new ResultTrain();
		}
		//判断是否需要处理列车数据
		if(originInfo.getUpdateTime() > originTrain.getUpdateTime()) 
		{
			String lineId = originInfo.getLineId();
			//根据线路获取车次窗列表
			HashMap<String, WindowInfo> windowRefMap = TrainServiceImpl.windowInfoMap.get(lineId); 
			//详细处理流程
			if(lineId.equals(KafkaMessageHelper.LINE_03))
			{
				//3、4号线共线段特殊处理，列车组编号以4起始时认定为4号线
				if (originInfo.getTrainGroupId().startsWith("4"))
				{
					lineId = KafkaMessageHelper.LINE_04;
					windowRefMap = TrainServiceImpl.windowInfoMap.get(lineId);
				}
			}
			//根据ATS消息中的车次窗获取对应的CELL与PLATFORM
			WindowInfo windowInfo = windowRefMap.get(originInfo.getWinHandle()); 
			if(windowInfo != null && !(windowInfo.getWindowRef().equals(originTrain.getPosition())))
			{
				try
				{
					originTrain.setLineId(lineId);
					originTrain.setPosType(windowInfo.getWindowType());
					if(StringUtils.isBlank(originInfo.getTrainGroupId()))
					{
						//列车编组号为空时，使用index作为编组号
						originTrain.setTrainGroupId(KafkaMessageHelper.TRAIN_GROUP_PREFIX + originInfo.getTrainIndex());
					}else {
						originTrain.setTrainGroupId(originInfo.getTrainGroupId());
					}
					originTrain.setTrainIndex(originInfo.getTrainIndex());
					originTrain.setUpdateTime(originInfo.getUpdateTime());
					originTrain.setPosition(windowInfo.getWindowRef()); //车次窗对应CELL或PLATFORM的字符串
					originTrain.setPosition2(windowInfo.getWindowRef2());
//					if(StringUtils.isBlank(originTrain.getUrl()) && TrainServiceImpl.iorUrlRefMap.containsKey(lineId))//添加一键观车URL
//					{
//						originTrain.setUrl(TrainServiceImpl.iorUrlRefMap.get(lineId).get(originTrain.getTrainGroupId()));
//					}
					if(StringUtils.isBlank(originTrain.getCarGroup()) || StringUtils.isBlank(originTrain.getCarType()))
					{
						if(TrainServiceImpl.iorUrlRefMap.containsKey(lineId) && TrainServiceImpl.iorUrlRefMap.get(lineId).containsKey(originTrain.getTrainGroupId()))
						{
							IorBean iorBean = TrainServiceImpl.iorUrlRefMap.get(lineId).get(originTrain.getTrainGroupId());
							originTrain.setCarGroup(iorBean.getTrainGroup());
							originTrain.setCarType(iorBean.getTrainType());
						}else {
							originTrain.setCarGroup(originTrain.getTrainGroupId());
							originTrain.setCarType(originTrain.getLineId() + "A01");
						}
					}
					processTrainLevel(originTrain); //处理车辆满载率等级
					//System.out.println(originTrain + " " +originTrain.getPosition() + " " + originTrain.getTrainGroupId());
					ATSMessageBean atsMessageBean = new ATSMessageBean();
					atsMessageBean.setType("update");
					atsMessageBean.setData(originTrain);
					ATSWebsocketController.broadCastInfo(MAPPER.writeValueAsString(atsMessageBean));
				} catch (JsonProcessingException e)
				{
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (IOException e)
				{
					// TODO Auto-generated catch block
					e.printStackTrace();
				}catch (Exception e) {
					// TODO: handle exception
					logger.info("exception occured:oringin info:"+originInfo);
					e.printStackTrace();
				}
			}else {
				//logger.info("ref:{},trainpos:{}",windowInfo.getWindowRef(),originTrain.getPosition());
			}
		}
		return originTrain;
	}
	
	private void processTrainLevel(ResultTrain originTrain)
	{
		@SuppressWarnings("unchecked")
		Map<String, Map<String, Integer>> trainFullLoad = (Map<String, Map<String, Integer>>) SpringContainer.getBean(AppCacheService.class).getCacheObject("trainFullLoad");
		if(trainFullLoad != null && trainFullLoad.containsKey(originTrain.getLineId()))
		{
			if(trainFullLoad.get(originTrain.getLineId()).containsKey(originTrain.getPosition()))
			{
				int level = trainFullLoad.get(originTrain.getLineId()).get(originTrain.getPosition());
				originTrain.setLevel(level);
			}
		}
	}
	
	private void removeTrain(TrainInfo trainInfo)
	{
		if(StringUtils.isNotBlank(trainInfo.getTrainIndex()))
		{
			SpringContainer.getBean(TrainService.class).removeTrainByTrainIndex(trainInfo.getTrainIndex(), trainInfo.getLineId());
		}
	}
	public String getWorkerName()
	{
		return workerName;
	}
	public void setWorkerName(String workerName)
	{
		this.workerName = workerName;
	}
}
